#include "config.h"
#include "common.h"
#include "cuckoo.h"
#include "arraylist.h"

typedef struct event_ptr
{
	int fd;
	char* send_buf;
	int len;
	int type;//0=client 1=datanodeasserver 2=datanodeasclient
}event_ptr;

DataNode* datanodes=NULL;
int** block_loc=NULL;
int nodeid;
struct epoll_event datanode_evs_asclient[DATANODE_NUM];
struct epoll_event datanode_evs_asserver[DATANODE_NUM];
int ep_fd;
long long id=0;
cuckoo* update_ctx_map=NULL;
cuckoo* update_entity_map=NULL;
char update_block[BLK_SIZE];
int parity[BLK_SIZE];

//====================================================
void process_datanode_updateCheck(event_ptr* ptr);
void send_WriteResponse(WriteResponse* response,int fd);
void send_ReadResponse(ReadResponse* response,int fd);
void recv_UpdateCheck(UpdateCheck* check,int fd);
void process_datanode_updateEntity(event_ptr* ptr);
void send_UpdateCheck(UpdateCheck* check,int fd);
void recv_UpdateEntity(UpdateEntity* entity,int fd);
void process_client_request(event_ptr* ptr);
void send_UpdateEntity(UpdateEntity* entity,int paritynodeid);
void recv_WriteRequest(WriteRequest* request,int fd);
void* conn2datanode(void* arg);

int main(int argc, char const *argv[])
{
	datanodes = read_config_Node(NODE_CONFIG);
	update_ctx_map = cuckoo_init(100);
	update_entity_map = cuckoo_init(100);
	nodeid = current_node(datanodes);

	struct epoll_event ev,events[65];

	ep_fd = epoll_create(256);//256其实没有什么意义，在最新的内核中可有可无
	int listen_fd = server_bind(PORT,20);

	ev.data.fd = listen_fd;
	ev.events=EPOLLIN;

	epoll_ctl(ep_fd,EPOLL_CTL_ADD,listen_fd,&ev);

	pthread_t conn2datanode_threadid;
	pthread_create(&conn2datanode_threadid,NULL,conn2datanode,NULL);

	while(1){
		int nfds = epoll_wait(ep_fd,events,65,-1);
		//printf("need to handle %d events\n",nfds);
		for (int i = 0; i < nfds; ++i)
		{
			if (events[i].data.fd == listen_fd)
			{
				int conn_fd = server_accept(listen_fd);
			
				struct sockaddr peer_addr={0};
				int peer_addr_len=sizeof(struct sockaddr);
				getpeername(conn_fd,&peer_addr,&peer_addr_len); 
				int peer_id = get_nodeid(&peer_addr,datanodes);
				
				if (peer_id==-1)
				{
					ev.data.fd=  conn_fd;
					ev.events = EPOLLIN;

					event_ptr* ptr = (event_ptr*)malloc(sizeof(event_ptr));
					ptr->len = 0;
					ptr->send_buf = (char*)malloc(BUFFER_SIZE);
					ptr->fd = conn_fd;
					ptr->type=0;
					ev.data.ptr = (void*)ptr;
					epoll_ctl(ep_fd,EPOLL_CTL_ADD,conn_fd,&ev);
				}else{
					datanode_evs_asserver[peer_id].data.fd = conn_fd;
					datanode_evs_asserver[peer_id].events = EPOLLIN;

					event_ptr* ptr = (event_ptr*)malloc(sizeof(event_ptr));
					ptr->len = 0;
					ptr->send_buf = (char*)malloc(BUFFER_SIZE);
					ptr->fd = conn_fd;
					ptr->type=1;
					datanode_evs_asserver[peer_id].data.ptr = (void*)ptr;
					epoll_ctl(ep_fd,EPOLL_CTL_ADD,conn_fd,&datanode_evs_asserver[peer_id]);
				}
				printf("peer_id:%d request to connect conn_fd:%d\n",peer_id,conn_fd);

			}else if(events[i].events&EPOLLIN){
				//解析接收到的WriteRequest	
				int fd = events[i].data.fd;
				event_ptr* ptr = (event_ptr*)events[i].data.ptr;
				if (ptr->type==0)
				{
					printf("process_client_request\n");
					process_client_request(ptr);
				}else if(ptr->type==1){
					printf("process_datanode_updateEntity\n");
					process_datanode_updateEntity(ptr);
				}else if(ptr->type==2){
					printf("process_datanode_updateCheck\n");
					process_datanode_updateCheck(ptr);
				}else{
					printf("unkown type fd\n");
				}

			}else if(events[i].events&EPOLLOUT){
				printf("write event is doing\n");
				event_ptr* ptr =(event_ptr*)events[i].data.ptr;
				memcpy(ptr->send_buf,&ptr->len,sizeof(int)); 
				send_bytes(ptr->fd,ptr->send_buf,ptr->len+sizeof(int));
				ptr->len=0;
				events[i].events = EPOLLIN;
				//events[i].data.fd = ptr->fd;
				epoll_ctl(ep_fd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
			}else{
				printf("the unkown event happened\n");
			}

		}
	}

	return 0;
}

void process_datanode_updateCheck(event_ptr* ptr)
{
	int len=0;
	char* int_buf = (char*)malloc(sizeof(int));
	recv_bytes(ptr->fd,int_buf,sizeof(int));
	memcpy(&len,int_buf,sizeof(int));
	free(int_buf);
	while(len>0){
		UpdateCheck* check = (UpdateCheck*)malloc(sizeof(UpdateCheck));
		recv_UpdateCheck(check,ptr->fd);
		
		char* key = convert_id_to_string(check->id);
	    UpdateContext* update_ctx =  (UpdateContext*)cuckoo_get(update_ctx_map,key);

	    if(check->res==1){
	      update_ctx->ready--;
	      if (update_ctx->ready==0)
	      {
	        UpdateEntity* entity = (UpdateEntity*)malloc(sizeof(UpdateEntity));
	        entity->id = update_ctx->id;
	        entity->type = 1;
	        for (int i = 0; i < R; ++i)
	        {
	          send_UpdateEntity(entity,update_ctx->request->paritynodeid[i]);
	        }
	        free(entity);
	      }
	    }

	    if (check->res==2)
	    {
	      update_ctx->status--;
	      if (update_ctx->status==0)
	      {
	        memcpy(update_block,update_ctx->request->newdata,BLK_SIZE);
	        WriteResponse* response = (WriteResponse*)malloc(sizeof(WriteResponse));
	        response->type=1;
	        response->responseid=update_ctx->request->requestid;
	        response->res=1;
	        send_WriteResponse(response,update_ctx->fd);
	        free(response);
	        cuckoo_remove(update_ctx_map,key);
	        free(update_ctx->request);
	        free(update_ctx);
	      } 
	    }
	    free(check);
    	free(key);
    	len=len-sizeof(UpdateCheck);
	}
    
}

void send_WriteResponse(WriteResponse* response,int fd)
{
	size_t needSend =sizeof(WriteResponse);
    char *sendBuf=(char*)malloc(needSend);
    memcpy(sendBuf,response,needSend); 
    send_bytes(fd,sendBuf,needSend);
    free(sendBuf);
}

void send_ReadResponse(ReadResponse* response,int fd)
{
	size_t needSend =sizeof(ReadResponse);
    char *sendBuf=(char*)malloc(needSend);
    memcpy(sendBuf,response,needSend); 
    send_bytes(fd,sendBuf,needSend);
    free(sendBuf);
}

void recv_UpdateCheck(UpdateCheck* check,int fd)
{
	size_t needRecv =sizeof(UpdateCheck);
    char *recvBuf=(char *)malloc(needRecv);
    recv_bytes(fd,recvBuf,needRecv);
    memcpy(check,recvBuf,needRecv);  //反序列化
    free(recvBuf);
}

void process_datanode_updateEntity(event_ptr* ptr)
{
	int len=0;
	char* int_buf = (char*)malloc(sizeof(int));
	recv_bytes(ptr->fd,int_buf,sizeof(int));
	memcpy(&len,int_buf,sizeof(int));
	free(int_buf);
	while(len>0){
		UpdateEntity* entity = (UpdateEntity*)malloc(sizeof(UpdateEntity));
		printf("recv_UpdateEntity\n");
		recv_UpdateEntity(entity,ptr->fd);

		printf("entity=%d\n",entity->type);
		if (entity->type==0){
			char* key = convert_id_to_string(entity->id);
	      	cuckoo_insert(update_entity_map,key,(void*)entity);

	      	UpdateCheck* check = (UpdateCheck*)malloc(sizeof(UpdateCheck));
	      	check->id = entity->id;
	      	check->res = 1;
	      	send_UpdateCheck(check,ptr->fd);

	      	free(key);
	      	free(check);
		}

		if (entity->type==1)
		{
			char* key = convert_id_to_string(entity->id);
			UpdateEntity* entity2 = (UpdateEntity*)cuckoo_get(update_entity_map,key);

			for (int i = 0; i < BLK_SIZE; ++i) parity[i] = parity[i]+entity2->newdata[i];
			UpdateCheck* check = (UpdateCheck*)malloc(sizeof(UpdateCheck));
			check->id = entity2->id;
			check->res = 2;
			send_UpdateCheck(check,ptr->fd);

			cuckoo_remove(update_entity_map,key);
			free(key);
			free(entity2);
			free(entity);
			free(check);
		}
		len=len-sizeof(UpdateEntity);
	}
}

void send_UpdateCheck(UpdateCheck* check,int fd)
{
	printf("send_UpdateCheck fd:%d\n",fd);
	int i;
	event_ptr* ptr=NULL;
	for (i = 0; i < DATANODE_NUM; i++)
	{
		if (i==nodeid) continue;
		ptr=(event_ptr*)datanode_evs_asserver[i].data.ptr;
		printf("fd:%d\n",ptr->fd);
		if (ptr->fd==fd) break;
	}
	printf("datanode_evs_asserver：%d\n",i);
	datanode_evs_asserver[i].events=EPOLLIN|EPOLLOUT;
	//datanode_evs_asserver[i].data.fd = ptr->fd;

	if (ptr->len+sizeof(UpdateCheck)+sizeof(int)>=BUFFER_SIZE)
	{
		memcpy(ptr->send_buf,&ptr->len,sizeof(int));
		send_bytes(ptr->fd,ptr->send_buf,ptr->len+sizeof(int));
		ptr->len = 0;
	}
	memcpy(ptr->send_buf+ptr->len+sizeof(int),check,sizeof(UpdateCheck));
	ptr->len = ptr->len+sizeof(UpdateCheck);
	epoll_ctl(ep_fd,EPOLL_CTL_MOD,ptr->fd,&datanode_evs_asserver[i]);

}

void recv_UpdateEntity(UpdateEntity* entity,int fd)
{
	size_t needRecv =sizeof(UpdateEntity);
    char *recvBuf=(char *)malloc(needRecv);
    recv_bytes(fd,recvBuf,needRecv);
    memcpy(entity,recvBuf,needRecv);  //反序列化
    free(recvBuf);
}

void recv_WriteRequest(WriteRequest* request,int fd)
{
	size_t needRecv =sizeof(WriteRequest);
    char *recvBuf=(char *)malloc(needRecv);
    recv_bytes(fd,recvBuf,needRecv);
    memcpy(request,recvBuf,needRecv);  //反序列化
    free(recvBuf);
}

void process_client_request(event_ptr* ptr)
{	

	size_t needRecv =sizeof(WriteRequest);
    char *recvBuf=(char *)malloc(needRecv);
    recv_bytes(ptr->fd,recvBuf,sizeof(int));

    int trace_type = 0;
    memcpy(&trace_type,recvBuf,sizeof(int));

    if (trace_type==0)
    {
    	recv_bytes(ptr->fd,recvBuf+sizeof(int),sizeof(ReadRequest)-sizeof(int));
    	ReadRequest* request = (ReadRequest*)malloc(sizeof(ReadRequest));
    	memcpy(request,recvBuf,sizeof(ReadRequest));  //反序列化
    	ReadResponse* response = (ReadResponse*)malloc(sizeof(ReadResponse));
    	response->type=0;
        response->responseid = request->requestid;
        response->res = 1;
        gene_radm_buff(response->newdata,BLK_SIZE);
        send_ReadResponse(response,ptr->fd);
        printf("send_ReadResponse finish\n");
        free(response);
        free(request);
        
    }else if (trace_type==1)
    {
    	recv_bytes(ptr->fd,recvBuf+sizeof(int),sizeof(WriteRequest)-sizeof(int));
    	WriteRequest* request = (WriteRequest*)malloc(sizeof(WriteRequest));
    	memcpy(request,recvBuf,sizeof(WriteRequest));  //反序列化
    	UpdateContext* update_ctx = (UpdateContext*)malloc(sizeof(UpdateContext));
		update_ctx->id =getcurrent_time()*10+id++;
		update_ctx->fd = ptr->fd;
		update_ctx->status = R;
		update_ctx->ready = R;
		update_ctx->request = request;

		char* key = convert_id_to_string(update_ctx->id);
		cuckoo_insert(update_ctx_map,key,(void*)update_ctx);
		free(key); 
		  
	    UpdateEntity* entity = (UpdateEntity*)malloc(sizeof(UpdateEntity));
	    entity->id = update_ctx->id;
	    for (int i = 0; i < BLK_SIZE; ++i) entity->newdata[i] = request->newdata[i]-update_block[i];

	    entity->type=0;
	    entity->blockid = request->blockid;
	    entity->checksum = CHECK_SUM;
	    for (int i = 0; i < R; ++i)
	    {
	      entity->parityid = request->parityid[i];    
	      send_UpdateEntity(entity,request->paritynodeid[i]);
	    }
	    free(entity);
    }else
    {
    	 printf("error trace_type:%d.\n",trace_type);
    }
    free(recvBuf);
}

void send_UpdateEntity(UpdateEntity* entity,int paritynodeid)
{
	//将内容写至缓冲区，修改为可写事件
	printf("send_UpdateEntity\n");
	struct epoll_event event = datanode_evs_asclient[paritynodeid];
	event.events = EPOLLOUT|EPOLLIN;
	event_ptr* ptr = (event_ptr*)event.data.ptr;

	//event.data.fd = ptr->fd;
	//printf("fd:%d\n", ptr->fd);
	if (ptr->len+sizeof(UpdateEntity)+sizeof(int)>=BUFFER_SIZE)
	{
		memcpy(ptr->send_buf,&ptr->len,sizeof(int));
		send_bytes(ptr->fd,ptr->send_buf,ptr->len+sizeof(int));
		ptr->len = 0;
	}	
	memcpy(ptr->send_buf+ptr->len+sizeof(int),entity,sizeof(UpdateEntity));
	ptr->len = ptr->len+sizeof(UpdateEntity);
	epoll_ctl(ep_fd,EPOLL_CTL_MOD,ptr->fd,&event);
}

void* conn2datanode(void* arg)
{
	sleep(3);
	for (int i = 0; i < DATANODE_NUM; ++i)
	{
		if (datanodes[i].nodeID!=nodeid)
		{
			int conn_fd = connect_try(datanodes[i].tcp_ip,PORT);
			datanode_evs_asclient[i].data.fd=conn_fd;
			datanode_evs_asclient[i].events= EPOLLIN;
			event_ptr* ptr = (event_ptr*)malloc(sizeof(event_ptr));
			ptr->fd = conn_fd;
			ptr->len = 0;
			ptr->send_buf = (char*)malloc(BUFFER_SIZE);
			ptr->type = 2;
			datanode_evs_asclient[i].data.ptr = (void*)ptr;
			epoll_ctl(ep_fd,EPOLL_CTL_ADD,conn_fd,&datanode_evs_asclient[i]);
		}
	}
	return NULL;
}